package cz.cvut.fel.loader;

import java.io.File;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.basho.riak.pbc.RequestMeta;
import com.basho.riak.pbc.RiakClient;
import com.basho.riak.pbc.RiakObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;
import com.mongodb.WriteConcern;

import cz.cvut.fel.email.Email;
import cz.cvut.fel.parser.JavaMailParser;
import cz.cvut.fel.parser.Parser;
import cz.cvut.fel.persister.RabbitMQPersister;
import cz.cvut.fel.persister.RiakPersister;
import cz.cvut.fel.stats.Statistics;

public class LoadJob extends Thread {
	final static Logger log = LoggerFactory.getLogger(LoadJob.class);
	private Parser parser = new JavaMailParser();
	private Statistics stats;
	private Persister in;
	private Persister out;
	private boolean stop = false;

	public LoadJob(Persister in, Persister out) {
		super();
		stats = Statistics.getInstance();
		this.in = in;
		this.out = out;
	}

	public void run() {
		log.info("Started thread with id: " + this.getId());

		while (!stop) { // blocked on input
			try {
				long start = System.currentTimeMillis(); // may be better after
															// reading
				// boolean success = false;
				byte[] data = in.readBytes();

				Email em = parser.parseEmail(data);
				out.store(em);
				stats.latency.getAndAdd(System.currentTimeMillis() - start);
				stats.docs.getAndIncrement();
			} catch (Exception e) {
				log.error(
						"Erro in thead: " + this.getId() + "->"
								+ e.getMessage(), e);
			}

			
		}

		log.info("Worker finished");
		Main.remaining.getAndDecrement();
		// String file = FileUtils.readFileToString(new File(path));
		// double x = Math.random();

	}

	/**
	 * @param parser
	 *            the parser to set
	 */
	public void setParser(Parser parser) {
		this.parser = parser;
	}

	/**
	 * @param in
	 *            the in to set
	 */
	public void setIn(Persister in) {
		this.in = in;
	}

	/**
	 * @return the in
	 */
	public Persister getIn() {
		return in;
	}

	/**
	 * @param out
	 *            the out to set
	 */
	public void setOut(Persister out) {
		this.out = out;
	}

	/**
	 * @return the out
	 */
	public Persister getOut() {
		return out;
	}

	/**
	 * @param stop
	 *            the stop to set
	 */
	public void setStop(boolean stop) {
		this.stop = stop;
	}

	/**
	 * @return the stop
	 */
	public boolean isStop() {
		return stop;
	}

	/*
	 * for (int p = 0; p < batchSize; p++) { objs[p] = new
	 * RiakObject("TESTING2", prefix + i + p, data); }
	 * 
	 * 
	 * c.store(objs, m);
	 * 
	 * 
	 * } else { if (db == 0) { obj = new RiakObject("TESTING2", prefix + i,
	 * data); c.store(obj, m); } else { //System.out.println(data.length);
	 * //bson = BSON.decode(data);
	 * 
	 * bson = new BasicDBObject ();
	 * 
	 * 
	 * 
	 * bson.put("_id", (this.add ? this.number.getAndIncrement() :
	 * this.number.getAndDecrement()) ); //bson.put("_id", UUID.randomUUID());
	 * 
	 * // bson.put("_id", prefixIn+ (id*10000000)+i);
	 * //((ran.nextInt()*10000000) //bson.put("sKey",
	 * splitter[i%3]+""+ran.nextInt()); //bson.put("_id", (""+splitter[i%3])+
	 * ((this.getId() % 2) == 1 ? Main.aiD.getAndIncrement() :
	 * Main.aiT.getAndDecrement()));
	 * 
	 * //prefixIn+Main.ai.getAndIncrement()); //bson.put("_id", this.getId()+
	 * ""+ i); // bson.put("my", i%2); //bson.put(key, v) prefix + i
	 * bson.put("data", data); // coll.insert(bson,WriteConcern.NORMAL); //
	 * System.out.println( bson.get("_id"));
	 * //mongo.getDB("").getCollection("").insert(arr)
	 * 
	 * } }
	 */
	/*
	 * private void storeToRiak(Email email) { if (email == null) {
	 * log.info("ODSTRANIT"); return; }
	 * 
	 * for (int i = 0; i < email.getAttachments().size(); i++) { try { /*
	 * RiakObject obj = null; if (email.getAttachments().size() > 1 &&
	 * email.getAttachments().size() == (i-1)) obj = new RiakObject("TEST1",
	 * email.getKeys().get(i),email.getAttachments ().get(i).getBytes()); else
	 * obj = new RiakObject("ATTACHMENT1",
	 * email.getKeys().get(i),email.getAttachments ().get(i).getBytes());
	 * 
	 * c.store(obj);
	 */
	/*
	 * mq.store("null", email.getAttachments().get(i).getBytes()); stats.add();
	 * // log.info("STORED "+email.getKeys().get(i)); } catch (Exception e) {
	 * log.error("Unable to store email " + email, e);
	 * 
	 * } }
	 * 
	 * log.debug(" Thead-ID: " + this.getId() + " storing file " + path);
	 * 
	 * }
	 */
}
